package com.fanxuankai.boot.mqbroker;

import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.BooleanUtil;
import com.fanxuankai.boot.mqbroker.autoconfigure.Clear;
import com.fanxuankai.boot.mqbroker.autoconfigure.MqBrokerProperties;
import com.fanxuankai.boot.mqbroker.consumer.MsgReceiveMapper;
import com.fanxuankai.boot.mqbroker.producer.MsgSendMapper;
import com.fanxuankai.boot.mqbroker.producer.autoconfigure.MqBrokerProducerProperties;
import com.fanxuankai.boot.redis.RedisUtils;
import com.fanxuankai.commons.util.OptionalUtils;
import org.springframework.core.env.Environment;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.CronTrigger;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 消息清理定时任务配置
 *
 * @author fanxuankai
 */
public class ClearMsgTaskConfigurer implements SchedulingConfigurer {
    @Resource
    private MqBrokerProperties properties;
    @Resource
    private MsgSendMapper sendMapper;
    @Resource
    private MsgReceiveMapper receiveMapper;
    @Resource
    private Environment environment;

    @Override
    public void configureTasks(@NonNull ScheduledTaskRegistrar scheduledTaskRegistrar) {
        if (properties.getClear().isEnabled()) {
            clearMsg(scheduledTaskRegistrar);
        }
    }

    private void clearMsg(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        Map<String, Set<String>> externalEventMap = getExternalEventMap();
        Clear clear = properties.getClear();
        long sendExpire = clear.getExpire() / 1000;
        // 接收表延长一天
        long receiveExpire = clear.getExpire() / 1000 + 86400;
        Set<String> externalEventNames = getExternalEventNames();
        String key = "lock:fxk:mq-broker:clearSendMsg:" + OptionalUtils.ofNullable(properties.getId())
                .orElse(environment.getProperty("spring.application.name"));
        scheduledTaskRegistrar.addTriggerTask(() -> {
            if (BooleanUtil.isTrue(RedisUtils.ValueOps.setIfAbsent(key, NetUtil.getLocalhostStr(),
                    Duration.ofMillis(clear.getTimeout())))) {
                // 先清理发送表
                sendMapper.clear(sendExpire, externalEventNames, clear.getExcludeEvents());
                if (!externalEventMap.isEmpty()) {
                    for (Map.Entry<String, Set<String>> entry : externalEventMap.entrySet()) {
                        sendMapper.clearExternal(sendExpire, entry.getKey(), entry.getValue(), clear.getExcludeEvents());
                    }
                }
                // 再清理接收表
                receiveMapper.clear(receiveExpire, clear.getExcludeEvents());
                RedisUtils.KeyOps.delete(key);
            }
        }, triggerContext -> new CronTrigger(clear.getCronExpression()).nextExecutionTime(triggerContext));
    }

    private Set<String> getExternalEventNames() {
        return properties.getExternalEvents()
                .stream()
                .map(MqBrokerProducerProperties.ExternalEvent::getName)
                .collect(Collectors.toSet());
    }

    private Map<String, Set<String>> getExternalEventMap() {
        return properties.getExternalEvents()
                .stream()
                .filter(externalEvent -> externalEvent.getSchema() != null)
                .collect(Collectors.groupingBy(MqBrokerProducerProperties.ExternalEvent::getSchema,
                        Collectors.mapping(MqBrokerProducerProperties.ExternalEvent::getName, Collectors.toSet())));
    }
}
